package com.twoandone.tms.realtime.app.dwd;

import com.twoandone.tms.realtime.utils.KafkaUtil;
import com.twoandone.tms.realtime.utils.MySqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

public class DwdServiceTaskDelivery {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        StreamTableEnvironment STE = StreamTableEnvironment.create(env);

        STE.executeSql(KafkaUtil.getTopicDbDDL("Dwd_Service_express_task_delivery"));
        STE.getConfig().setIdleStateRetention(Duration.ofSeconds(15 * 60 + 5));
        Table deliveryTable = STE.sqlQuery("select\n" +
                "     `data`['id'] id,\n" +
                "     `data`['order_id'] order_id,\n" +
                "     `data`['status'] status ,\n" +
                "     `data`['org_id'] org_id,\n" +
                "     `data`['org_name'] org_name,\n" +
                "     `data`['courier_emp_id'] courier_emp_id,\n" +
                "     `data`['courier_name'] courier_name,\n" +
                "     `data`['estimated_end_time'] estimated_end_time,\n" +
                "     `data`['start_delivery_time'] start_delivery_time ,\n" +
                "     `data`['delivered_time'] delivered_time,\n" +
                "     `data`['is_rejected'] is_rejected,\n" +
                "     `data`['create_time'] create_time,\n" +
                "     `data`['update_time'] update_time,\n" +
                "     `data`['is_deleted'] is_deleted,\n" +
                "     proc_time,\n" +
                "     ts\n" +
                "     from topic_db\n" +
                "     where `table` = 'express_task_delivery'\n" +
                "     and `type` = 'insert' or  `type`  =  'update'");
        STE.createTemporaryView("expressDelivery", deliveryTable);
        // STE.executeSql("select status from delivery").print();

        // 建立字典表
        STE.executeSql(MySqlUtil.getBaseDicLookUpDDL());
        // // STE.executeSql("select * from base_dic").print();

        //lookupjoin
        Table lookUpJoin = STE.sqlQuery("SELECT \n" +
                "                        `delivery`.id,\n" +
                "                        `delivery`.order_id,\n" +
                "                        `delivery`.status,\n" +
                "                        `delivery`.org_id,\n" +
                "                        `delivery`.org_name,\n" +
                "                        `delivery`.courier_emp_id,\n" +
                "                        `delivery`.courier_name,\n" +
                "                        `delivery`.estimated_end_time,\n" +
                "                        `delivery`.start_delivery_time,\n" +
                "                        `delivery`.delivered_time,\n" +
                "                        `delivery`.is_rejected,\n" +
                "                        `delivery`.create_time,\n" +
                "                        `delivery`.update_time,\n" +
                "                        `delivery`.is_deleted,\n" +
                "                        dic.name status_name  \n" +
                "                        FROM expressDelivery AS `delivery` JOIN base_dic FOR SYSTEM_TIME AS OF `delivery`.proc_time AS dic ON CAST(`delivery`.status AS bigint) = dic.id");
        STE.createTemporaryView("lookUpJoin", lookUpJoin);


        STE.executeSql("  CREATE TABLE dwd_service_express_delivery (\n" +
                "                        id string,\n" +
                "                        order_id string,\n" +
                "                        status string,\n" +
                "                        org_id string,\n" +
                "                        org_name string,\n" +
                "                        courier_emp_id string,\n" +
                "                        courier_name string,\n" +
                "                        estimated_end_time string,\n" +
                "                        start_delivery_time string,\n" +
                "                        delivered_time string,\n" +
                "                        is_rejected string,\n" +
                "                        create_time string,\n" +
                "                        update_time string,\n" +
                "                        is_deleted string,\n" +
                "                        status_name string,\n" +
                "                        PRIMARY KEY (id) NOT ENFORCED )"
                + KafkaUtil.getUpsertKafkaDDL("dwd_service_express_delivery"));

        //写入kafka
        STE.executeSql("insert into dwd_service_express_delivery select * from lookUpJoin");
    }
}
